package com.cetc.sdp.kmga.cs.stream;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.cetc.sdp.kmga.cs.device.*;
import com.cetc.sdp.kmga.cs.jdbc.DeviceDAO;
import com.cetc.sdp.kmga.cs.util.*;
import com.rabbitmq.client.QueueingConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.*;
import org.apache.spark.streaming.rabbitmq.RabbitMQUtils;
import org.apache.spark.streaming.rabbitmq.distributed.JavaRabbitMQDistributedKey;
import org.apache.spark.streaming.rabbitmq.models.ExchangeAndRouting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.spark_project.jetty.util.ConcurrentHashSet;
import scala.Tuple2;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @description: 处理设备状态日志, Spark Streaming 实时流处理
 * @author： DengQiang
 * @date: 2017/8/17 16:01
 */
public class StateStream {
    /**
     * 设备离线
     */
    private static final int DEVICE_OFFLINE_STAT = 0x00;
    /**
     * 设备正常
     */
    private static final int DEVICE_ONLINE_STAT = 0x02;
    /**
     * 设备异常
     */
    private static final int DEVICE_EXCEPTION_STAT = 0x01;

    private static final Logger LOGGER = LoggerFactory.getLogger(StateStream.class);
    private static final int ARGS_LEN = 2;

    public static void main(String[] args) throws InterruptedException, IOException {
        if (args.length < ARGS_LEN) {
            System.out.println("Usage：<Application Name> <Batch duration> [Repartition] [Num of partition]");
            System.exit(1);
        }
        String appName = args[0];
        int duration = Integer.parseInt(args[1]);
        boolean repartiton = false;
        int numOfPar = 2;
        if (args.length >= 4) {
            repartiton = Boolean.valueOf(args[2]);
            numOfPar = Integer.valueOf(args[3]);
        }
        boolean useKafkaRecv = "kafka".equals(args[4]);
        boolean useRabbitmqRecv = "rabbitmq".equals(args[4]);

        StreamingConfiguration streamingConfiguration = new StreamingConfiguration("state-config.xml");

        String topic = streamingConfiguration.getAppProperties().get("kafka.stream.topic");
        String maxOffCountStr = streamingConfiguration.getAppProperties().getOrDefault("max.offline.count", "3");
        int maxOfflineCount = Integer.parseInt(maxOffCountStr);
        boolean enableAutoBroadcast = Boolean.parseBoolean(streamingConfiguration.getAppProperties().getOrDefault("enable.auto.broadcast", "false"));
        int broadcastUpdateInterval = Integer.parseInt(streamingConfiguration.getAppProperties().getOrDefault("broadcast.interval.ms", "3600000"));

//        String mqHost = streamingConfiguration.getRabbitmqProperties().get("rabbitmq.host");
//        int mqPort = Integer.parseInt(streamingConfiguration.getRabbitmqProperties().get("rabbitmq.port"));
        String mqUsername = streamingConfiguration.getRabbitmqProperties().get("rabbitmq.username");
        String mqPassword = streamingConfiguration.getRabbitmqProperties().get("rabbitmq.password");
        String mqDevExchange = streamingConfiguration.getRabbitmqProperties().get("rabbitmq.dev.exchange");
        String mqNodeList = streamingConfiguration.getRabbitmqProperties().get("rabbitmq.node.list");
        String statQueues = streamingConfiguration.getRabbitmqProperties().get("stat.queue.list");

        Map<String, String> rabbitmqParams = new HashMap<>(32);
        rabbitmqParams.put("hosts", mqNodeList);
        rabbitmqParams.put("vHost", "/");
        rabbitmqParams.put("userName", mqUsername);
        rabbitmqParams.put("password", mqPassword);
        rabbitmqParams.put("ackType", "auto");
        rabbitmqParams.put("maxReceiveTime", args[5]);
        rabbitmqParams.put("maxMessagesPerPartition", args[6]);


        //初始化上下文环境
        SparkConf conf = new SparkConf().setAppName(appName);
        //使用Kryo序列化库
        conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
        //在Kryo序列化库中注册自定义的类集合
        conf.set("spark.kryo.registrator", StateKryoRegistrator.class.getName());
        JavaSparkContext jsc = new JavaSparkContext(conf);
        jsc.setLogLevel("ERROR");
        JavaStreamingContext jssc = new JavaStreamingContext(jsc, Durations.seconds(duration));

        //设置kafka参数
        Map<String, Object> kafkaParams = new HashMap<>(streamingConfiguration.getKafkaProperties());
        //初始化设备数据
        List<Tuple2<String, Integer>> allDevices = DeviceDAO.findAllDevice();
        Map<String, DeviceState> devStateMap = new ConcurrentHashMap<>(1024);
        List<DeviceExpEvent> hasOfflineDevList = DeviceDAO.findAllOfflineDevices();
        Map<String, DeviceExpEvent> hasOfflineDevMap = new ConcurrentHashMap<>(16);
        for (DeviceExpEvent event : hasOfflineDevList) {
            hasOfflineDevMap.put(event.getDeviceId(), event);
        }
        Set<String> initOffineDevSet = new ConcurrentHashSet<>();
        Set<String> devs = new HashSet<>();
        for (Tuple2<String, Integer> dev : allDevices) {
            if (hasOfflineDevMap.get(dev._1()) == null && dev._2() == DEVICE_ONLINE_STAT) {
                devStateMap.put(dev._1(), new DeviceState(dev._1(), false, maxOfflineCount));
            } else {
                devStateMap.put(dev._1(), new DeviceState(dev._1(), false, 0));
            }
            if (dev._2() == DEVICE_OFFLINE_STAT && hasOfflineDevMap.get(dev._1()) == null) {
                initOffineDevSet.add(dev._1());
            }
            devs.add(dev._1());
        }

        DeviceSetBroadcastWrapper deviceSetBroadcastWrapper = DeviceSetBroadcastWrapper.getInstance(jsc, devs);
        deviceSetBroadcastWrapper.setUpdateByFixRate(enableAutoBroadcast);
        deviceSetBroadcastWrapper.setInterval(broadcastUpdateInterval);

        //启动消息接收线程
        RabbitMQConsumer consumer = new RabbitMQConsumer();
        String[] nodes = mqNodeList.split(",");
        for (String node : nodes) {
            String[] ip = node.split(":");
            if (ip.length == 2) {
                consumer.addAddress(ip[0], Integer.parseInt(ip[1]));
            }
        }
        consumer.setXt(RabbitMQProducer.XT.FANOUT);
        consumer.setExchange(mqDevExchange);
        consumer.setUserName(mqUsername);
        consumer.setPassword(mqPassword);
        consumer.addMsgListener(message -> {
            System.out.println("Recv: " + message);
            DevNotify notify = JSON.parseObject(message, DevNotify.class);
            //设备添加标识
            String add = "Add";
            //设备删除标识
            String del = "Del";
            if (notify.getDevNums() != null) {
                if (add.equalsIgnoreCase(notify.getType())) {
                    for (String dev : notify.getDevNums()) {
                        //初始时假设设备状态为离线
                        deviceSetBroadcastWrapper.addDevice(dev);
                        devStateMap.put(dev, new DeviceState(dev, false, 0));
                        initOffineDevSet.add(dev);
                    }
                } else if (del.equalsIgnoreCase(notify.getType())) {
                    for (String dev : notify.getDevNums()) {
                        deviceSetBroadcastWrapper.removeDevice(dev);
                        devStateMap.remove(dev);
                    }
                }
            }
        });
        consumer.start();

        JavaInputDStream<ConsumerRecord<Object, Object>> kafkaStream = null;
        JavaDStream<String> dStream = null;
        if (useRabbitmqRecv) {
            List<JavaRabbitMQDistributedKey> distributedKeys = new LinkedList<>();
            String[] queues = statQueues.split(",");
            for (String queue : queues) {
                distributedKeys.add(new JavaRabbitMQDistributedKey(queue.trim(),
                        new ExchangeAndRouting(""),
                        rabbitmqParams));
            }
            Function<QueueingConsumer.Delivery, String> messageHandler = (Function<QueueingConsumer.Delivery, String>) message -> new String(message.getBody());
            dStream = RabbitMQUtils.createJavaDistributedStream(jssc, String.class, distributedKeys, rabbitmqParams, messageHandler);
        } else {
            kafkaStream = KafkaUtils.createDirectStream(
                    jssc,
                    LocationStrategies.PreferConsistent(),
                    ConsumerStrategies.Subscribe(Arrays.asList(topic), kafkaParams)
            );
            dStream = kafkaStream.map(r -> r.value().toString());
        }

        //clean stream
        JavaDStream<StateLog> cleanStream = (repartiton ? dStream.repartition(numOfPar) : dStream)
                .map(StateStream::parseDevStatLog)
                .filter(log -> log.getDevId() != null).cache();
        DeviceExpCheckBroadcastWrapper.getInstance(jsc).setUpdateByFixRate(enableAutoBroadcast);
        DeviceExpCheckBroadcastWrapper.getInstance(jsc).setInterval(broadcastUpdateInterval);
        cleanStream.transform(stream -> {
            Broadcast<DeviceExpCheck> bc = DeviceExpCheckBroadcastWrapper.getInstance(jsc).updateAndGet();
            Broadcast<Set<String>> devsBc = DeviceSetBroadcastWrapper.getInstance(jsc, devs).updateAndGet();
            return stream.filter(l -> devsBc.getValue().contains(l.getDevId()))
                    .map(l -> new Tuple2<>(l, bc.getValue().check(l)));
        }).foreachRDD(javaRDD -> {
            javaRDD.foreachPartition(x -> {
                /**
                 * 保存异常信息：
                 * 1、首先加载活动的异常信息
                 * 2、比较找出已结束异常
                 * 3、添加新的异常
                 */
                Map<String, List<DeviceExpEvent>> activeExpEvent = DeviceDAO.findAllActiveExpEvent();
                List<StateLog> logs = new ArrayList<>();
                String now = LocalDateTime.now(Tool.zoneId).format(Tool.DEFAULT_TIME_FORMATTER);
                //新增异常
                Set<DeviceExpEvent> toExpDevice = new HashSet<>(128);

                while (x.hasNext()) {
                    Tuple2<StateLog, List<DeviceExpEvent>> tuple2 = x.next();
                    StateLog log = tuple2._1();
                    logs.add(log);
                    List<DeviceExpEvent> expEvents = tuple2._2();
                    if (expEvents.isEmpty()) {
                        //无异常
                        List<DeviceExpEvent> tmp = activeExpEvent.get(log.getDevId());
                        if (tmp != null && !tmp.isEmpty()) {
                            //添加异常结束时间
                            ThreadPool.getExecutorService().submit(() -> {
                                for (DeviceExpEvent expEvent : tmp) {
                                    expEvent.setEndTime(now);
                                    DeviceDAO.updateDeviceExpEvent(expEvent);
                                    DeviceDAO.updateDeviceStatus(expEvent.getDeviceId(), DEVICE_ONLINE_STAT);
                                }
                            });
                        }
                    } else {
                        //有异常
                        List<DeviceExpEvent> tmp = activeExpEvent.get(log.getDevId());
                        if (tmp == null || tmp.isEmpty()) {
                            //添加新异常
                            toExpDevice.addAll(expEvents);
                        } else {
                            boolean[] expAliveLookup = new boolean[tmp.size()];
                            for (DeviceExpEvent expEvent : expEvents) {
                                boolean flag = true;
                                int i = 0;
                                for (DeviceExpEvent oldExpEvent : tmp) {
                                    if (oldExpEvent.getExcpCode() == expEvent.getExcpCode()) {
                                        expAliveLookup[i] = true;
                                        flag = false;
                                        break;
                                    }
                                    i++;
                                }
                                if (flag) {
                                    toExpDevice.add(expEvent);
                                }
                            }
                            for (int i = 0; i < expAliveLookup.length; i++) {
                                if (!expAliveLookup[i]) {
                                    tmp.get(i).setEndTime(now);
                                    DeviceDAO.updateDeviceExpEvent(tmp.get(i));
                                }
                            }


                        }
                    }
                }
                //保存新增异常
                ThreadPool.getExecutorService().submit(() -> {
                    for (DeviceExpEvent expEvent : toExpDevice) {
                        DeviceDAO.saveDeviceExpEvent(expEvent);
                        DeviceDAO.updateDeviceStatus(expEvent.getDeviceId(), DEVICE_EXCEPTION_STAT);
                    }
                });
                //保存状态日志到数据库
                ThreadPool.getExecutorService().submit(() -> DeviceDAO.saveDeviceStateLog(logs, now));
            });
        });

        cleanStream.map(StateLog::getDevId)
                .foreachRDD(mapJavaRDD -> {
                    Broadcast<DeviceExpCheck> bc = DeviceExpCheckBroadcastWrapper.getInstance(jsc).updateAndGet();
                    DeviceExpCode offlineExpCode = bc.getValue().getOfflineExpCode();

                    String time = LocalDateTime.now(Tool.zoneId).format(Tool.DEFAULT_TIME_FORMATTER);
                    List<String> onlineDevs = mapJavaRDD.collect();
                    //从离线中恢复的设备
                    List<DeviceExpEvent> backToOnlineDevs = new ArrayList<>(32);
                    List<String> needUpdateInitStatDevs = new ArrayList<>(16);
                    for (String devId : onlineDevs) {
                        DeviceState ds = devStateMap.get(devId);
                        if (ds != null) {
                            ds.setCount(maxOfflineCount);
                            ds.setUpdate(true);
                        } else {
                            devStateMap.put(devId, new DeviceState(devId, true, maxOfflineCount));
                        }

                        DeviceExpEvent expEvent = hasOfflineDevMap.remove(devId);
                        if (expEvent != null) {
                            expEvent.setEndTime(time);
                            backToOnlineDevs.add(expEvent);
                        }
                        if (!initOffineDevSet.isEmpty() && initOffineDevSet.remove(devId)) {
                            needUpdateInitStatDevs.add(devId);
                        }
                    }
                    //从在线进入离线的设备
                    List<DeviceExpEvent> toOfflineDevs = new ArrayList<>();
                    for (Map.Entry<String, DeviceState> entry : devStateMap.entrySet()) {
                        String dev = entry.getKey();
                        DeviceState ds = entry.getValue();
                        if (!ds.isUpdate()) {
                            int count = ds.getCount() - 1;
                            if (count == 0) {
                                ds.setCount(count);
                                DeviceExpEvent offlineExpEvent = new DeviceExpEvent();
                                offlineExpEvent.setExcpCode(offlineExpCode.getExcpCode());
                                offlineExpEvent.setStartTime(time);
                                offlineExpEvent.setDeviceId(dev);
                                offlineExpEvent.setLevel(offlineExpCode.getLevel());

                                toOfflineDevs.add(offlineExpEvent);
                                hasOfflineDevMap.put(dev, offlineExpEvent);
                            } else if (count < 0) {
                                if (!hasOfflineDevMap.containsKey(dev)) {
                                    DeviceExpEvent offlineExpEvent = new DeviceExpEvent();
                                    offlineExpEvent.setExcpCode(offlineExpCode.getExcpCode());
                                    offlineExpEvent.setStartTime(time);
                                    offlineExpEvent.setDeviceId(dev);
                                    offlineExpEvent.setLevel(offlineExpCode.getLevel());

                                    toOfflineDevs.add(offlineExpEvent);
                                    hasOfflineDevMap.put(dev, offlineExpEvent);
                                }
                            } else {
                                ds.setCount(count);
                            }
                        } else {

                        }

                        //reset state
                        ds.setUpdate(false);
                    }

                    ThreadPool.getExecutorService().submit(() -> {
                        backToOnlineDevs.forEach(event -> {
                            DeviceDAO.updateDeviceExpEvent(event);
                            DeviceDAO.updateDeviceStatus(event.getDeviceId(), DEVICE_ONLINE_STAT);
                        });
                        needUpdateInitStatDevs.forEach(devId -> DeviceDAO.updateDeviceStatus(devId, DEVICE_ONLINE_STAT));
                        //当设备离线时，该设备的其余告警事件结束, 并更新设备状态
                        List<DeviceExpEvent> endExpEvent = new ArrayList<>();
                        for (DeviceExpEvent event : toOfflineDevs) {
                            endExpEvent.addAll(DeviceDAO.findActiveExpEventByDevId(event.getDeviceId()));
                            DeviceDAO.updateDeviceStatus(event.getDeviceId(), DEVICE_OFFLINE_STAT);
                        }
                        endExpEvent.forEach(event -> {
                            event.setEndTime(time);
                            DeviceDAO.updateDeviceExpEvent(event);
                        });
                        DeviceDAO.saveDeviceExpEvent(toOfflineDevs);
                    });
                });

        if (kafkaStream != null) {
            CanCommitOffsets offsets = (CanCommitOffsets)kafkaStream.inputDStream();
            kafkaStream.foreachRDD(javaRDD -> {
                OffsetRange[] offsetRanges = ((HasOffsetRanges) javaRDD.rdd()).offsetRanges();
                offsets.commitAsync(offsetRanges);
            });
        }

        jssc.start();
        jssc.awaitTermination();
        jssc.close();
    }

    /**
     * 解析设备状态日志
     * @param logStr
     * @return
     */
    public static StateLog parseDevStatLog(String logStr) {
        StateLog log = new StateLog();
        try {
            Map<String, Object> logMap = (Map<String, Object>) JSON.parse(logStr);
            log.setDevId((String) logMap.get("devId"));
            String logTypeStr = logMap.get("logType").toString();
            if (logTypeStr != null && !logTypeStr.isEmpty()) {
                log.setLogType(Integer.parseInt(logTypeStr));
            }
            JSONArray jsonArray = (JSONArray) logMap.get("datas");
            if (jsonArray != null && jsonArray.size() >= 6) {
                String cpuStr = (String) jsonArray.get(0);
                if (cpuStr != null && !cpuStr.isEmpty()) {
                    if (cpuStr.endsWith("%")) {
                        cpuStr = cpuStr.substring(0, cpuStr.length() - 1);
                    }
                    log.setCpuUse((int)Double.parseDouble(cpuStr));
                }

                String memUseStr = (String) jsonArray.get(1);
                if (memUseStr != null && !memUseStr.isEmpty()) {
                    String[] memArr = memUseStr.split("[/]");
                    if (memArr.length == 2) {
                        log.setMemUse((int)Double.parseDouble(memArr[0]));
                        log.setMemTotal((int)Double.parseDouble(memArr[1]));
                    }
                }

                String diskUseStr = (String) jsonArray.get(2);
                if (diskUseStr != null && !diskUseStr.isEmpty()) {
                    String[] diskArr = diskUseStr.split("[/]");
                    if (diskArr.length == 2) {
                        log.setDiskUse((int)Double.parseDouble(diskArr[0]));
                        log.setDiskTotal((int)Double.parseDouble(diskArr[1]));
                    }
                }

                String netStr = (String) jsonArray.get(3);
                if (netStr != null && !netStr.isEmpty()) {
                    String[] netArr = netStr.split("[/]");
                    if (netArr.length == 2) {
                        log.setNetRecv((int)Double.parseDouble(netArr[0]));
                        log.setNetSend((int)Double.parseDouble(netArr[1]));
                    }
                }

                String statStr = (String) jsonArray.get(4);
                log.setDevStatus(Integer.parseInt(statStr));

                log.setDevStatReason((String) jsonArray.get(5));
            }

        } catch (Exception e) {
            LOGGER.error("状态日志不合法, {}", logStr);
            e.printStackTrace();
        }
        return log;
    }

    private static class DeviceExpCheckBroadcastWrapper extends AbstractBroadcastWrapper<DeviceExpCheck> implements MessageListener {

        private volatile static DeviceExpCheckBroadcastWrapper INSTANCE;

        private DeviceExpCheckBroadcastWrapper(JavaSparkContext jsc) {
            super(jsc);
        }

        static DeviceExpCheckBroadcastWrapper getInstance(JavaSparkContext jsc) {
            if (INSTANCE == null) {
                synchronized (DeviceExpCheckBroadcastWrapper.class) {
                    if (INSTANCE == null) {
                        INSTANCE = new DeviceExpCheckBroadcastWrapper(jsc);
                    }
                }
            }
            return INSTANCE;
        }

        @Override
        protected Broadcast<DeviceExpCheck> broadcast(JavaSparkContext jsc) {
            return jsc.broadcast(new DeviceExpCheck(DeviceDAO.findAllDeviceExpCode()));
        }

        @Override
        public void fireMessage(String message) {
            System.out.println("Recv :" + message);
            refreshBroadcast();
        }
    }

}
